package scg.utils;

import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;

import org.gridgain.grid.GridException;
import org.gridgain.grid.GridFactory;
import org.gridgain.grid.GridJob;
import org.gridgain.grid.GridJobAdapter;
import org.gridgain.grid.GridJobResult;
import org.gridgain.grid.GridNode;
import org.gridgain.grid.GridTask;
import org.gridgain.grid.GridTaskAdapter;

import scg.utils.policies.loadbalancer.BalancerPolicy;


/**
 * Esta clase implementa la interface del middleware GridGain denominada {@link GridTask}. 
 * Su creador recibe como parámetro un entero, el cual, si es mayor a 1 indica que la tarea es 
 * no terminal y la generación de las tareas del siguiente nivel, estará a cargo del nodo
 * local desde donde se lanzó la aplicación. Por el contrario, si el entero es igual a 1,
 * se trata de una tarea terminal y es recién ahí donde entrará en juego el balanceador de carga que
 * haya sido configurado. *
 * @param <T> tipo de retorno de {@link Callable} t.
 */
public class NonTerminalTask<T> extends GridTaskAdapter<Callable<T>, T> {
    
	
	/**nivel recursivo de la tarea mediante el cual se decide su destino en el grid.
	 * si contiene el valor uno, se trata de una tarea terminal y su destino dependerá de la política
	 * de balanceo configurada. Por el contrario, si el valor es mayor a uno, su destino será el nodo
	 * local*/
	int recLevel;

	/**balanceador de carga que será utilizado para mapear las tareas terminales*/
	static BalancerPolicy balancer = null;
    
	/**@param el balanceador de carga que se utilizará para distribuir las tareas terminales
	 * */
	public static void setBalancerPolicy(BalancerPolicy bp){
		balancer = bp;
	}
	
	/**Contructor de la tarea
	 * @param nivel recursivo de la tarea mediante el cual se decide su destino en el grid.*/
	public NonTerminalTask(int recLevel){		
		this.recLevel = recLevel;		
	}
	
	
    protected Collection<? extends GridJob> split(int gridSize, Callable<T> arg) throws GridException {
        return Collections.singletonList(new GridJobAdapter<Serializable>((Serializable)arg) {
            /*
            * Simply execute command passed into the job and
            * returns result.
            */
            @SuppressWarnings("unchecked")
            public Serializable execute() throws GridException {
                final Callable<T> call = (Callable<T>)getArgument();

                // Execute command.
                try {
                    T res = call.call();

                    if (res != null && res instanceof Serializable == false) {
                        throw (GridException)new GridException("Result of callable must be Serializable: " + res).setData(60, "src/java/org/gridgain/grid/kernal/executor/GridExecutorCallableTask.java");
                    }

                    return (Serializable)res;
                }
                catch (RuntimeException e) {
                    throw e;
                }
                catch (Error e) {
                    throw e;
                }
                catch (Exception e) {
                    throw (GridException)new GridException("Failed to execute command.", e).setData(72, "src/java/org/gridgain/grid/kernal/executor/GridExecutorCallableTask.java");
                }
            }
        });
    }

    /**
     * {@inheritDoc}
     */
    @SuppressWarnings({"unchecked"})
    public T reduce(List<GridJobResult> results) throws GridException {
        assert results != null : "ASSERTION [line=83, file=src/java/org/gridgain/grid/kernal/executor/GridExecutorCallableTask.java]";
        assert results.size() == 1 : "ASSERTION [line=84, file=src/java/org/gridgain/grid/kernal/executor/GridExecutorCallableTask.java]";

        return (T)results.get(0).getData();
    }

    
    /**Este método realiza el mapeo de tareas a nodos dependiendo del nivel recursivo será el destino de
     * los trabajos que aloja dicha tarea*/
    public Map<? extends GridJob, GridNode> map(List<GridNode> subgrid, Callable<T> arg) throws GridException {
        assert subgrid != null : "ASSERTION [line=108, file=src/java/org/gridgain/grid/GridTaskSplitAdapter.java]";
        assert subgrid.isEmpty() == false : "ASSERTION [line=109, file=src/java/org/gridgain/grid/GridTaskSplitAdapter.java]";

        Collection<? extends GridJob> jobs = split(subgrid.size(), arg);

        if (jobs == null || jobs.isEmpty() == true) {
            throw (GridException)new GridException("Split returned no jobs.").setData(114, "src/java/org/gridgain/grid/GridTaskSplitAdapter.java");
        }

        Map<GridJob, GridNode> map = new HashMap<GridJob, GridNode>(jobs.size());

        for (GridJob job : jobs) {
        	if (recLevel > 1){//significa que es un trabajo intermedio entonces sigo creando trabajos en el nodo local.
        		map.put(job, GridFactory.getGrid().getLocalNode());        		
        	}
        	else{        		
        		map.put(job, balancer.getBalancedNode()); 
        	}        		
        }

        return map;
    }
		
}
